package com.xujianlong.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

public class MyKafkaUtil {
   private static final String KAFKA_SERVER="hadoop102:9092,hadoop103:9092,hadoop104:9092";

   private static Properties properties=new Properties();

   private static final String DEFAULT_TOPIC="DWD_DEFAULT_TOPIC";

   static{
       properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFKA_SERVER);
   }

   public static FlinkKafkaProducer<String> getKafkaSink(String topic){
       return new FlinkKafkaProducer<String>(topic,new SimpleStringSchema(),properties);
    }

    public static <T> FlinkKafkaProducer<T> getKafkaSink(KafkaSerializationSchema<T> KafkaSerializationSchema){
       //Default_Topic是要求你会写主题，但是你没写有个默认值
        //如果你没有CK FlinkKafkaProducer.Semantic 里面的代码会走None,即使你设置了EXACTLY_ONCE，端到端的精准一次性的木桶效应
       return new FlinkKafkaProducer<T>(DEFAULT_TOPIC,
               KafkaSerializationSchema,
               properties,
               FlinkKafkaProducer.Semantic.EXACTLY_ONCE
       );
    }

    //SimpleStringSchema 如果数据为Null会报错
    public static FlinkKafkaConsumer<String> getKafkaSource(String topic,String groupId){
       properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
       return new FlinkKafkaConsumer<String>(topic,
               new SimpleStringSchema(),
               properties);
    }

    public static FlinkKafkaConsumer<String> getKafkaSource1(String topic,String groupId){
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        return new FlinkKafkaConsumer<String>(topic,
                new KafkaDeserializationSchema<String>() {
                    @Override
                    public boolean isEndOfStream(String s) {
                        return false;
                    }
                    @Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
                       if(consumerRecord ==null || consumerRecord.value()==null){
                           //返回Null或者空串 下游过滤掉
                           return null;
                       }
                        return new String(consumerRecord.value());
                    }
                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                },
                properties);
    }

    //拼接Kafka相关属性到DDL
    public static String getKafkaDDL(String topic, String groupId) {
        return "'connector' = 'kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                "  'format' = 'json', " +
                "  'scan.startup.mode' = 'latest-offset'";
    }
}
